package com.netty.future;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.netty.call.AsyncRPCCallback;
import com.netty.client.ClientProxy;
import com.netty.common.protocol.RpcRequest;
import com.netty.common.protocol.RpcResponse;

public class RPCFuture implements Future<Object>{
	
	private static final Logger logger = LoggerFactory.getLogger(RPCFuture.class);
	
	private Sync sync;
	private RpcRequest request;
	private RpcResponse response;
	private long startTime;
	/**响应时间阀值*/
	private long responseTimeThreshold = 5000;
	private List<AsyncRPCCallback> pendingCallbacks = new ArrayList<AsyncRPCCallback>();
	private ReentrantLock lock = new ReentrantLock();
	
	
	public RPCFuture(RpcRequest request) {
		this.sync = new Sync();
		this.request = request;
		this.startTime = System.currentTimeMillis();
	}

	/**
	 * 关闭操作
	 */
	@Override
	public boolean cancel(boolean mayInterruptIfRunning) {
		throw new UnsupportedOperationException();
	}

	/**
	 * 取消操作
	 */
	@Override
	public boolean isCancelled() {
		throw new UnsupportedOperationException();
	}

	/**
	 * 请求释放完成
	 */
	@Override
	public boolean isDone() {
		return sync.isDone();
	}

	@Override
	public Object get() throws InterruptedException, ExecutionException {
		sync.acquire(-1);
		if (this.response != null) {
			return this.response.getResult();
		}
		return null;
	}

	/**
	 * 完成请求
	 * @param response
	 */
	public void done(RpcResponse response){
		this.response = response;
		sync.release(1);
		invokeCallbacks();
		//响应时间
		long responseTime = System.currentTimeMillis()-startTime;
		if (responseTime > this.responseTimeThreshold) {
			logger.warn("服务响应时间慢，请求id=" + response.getRequestId() + ". 响应时间= " + responseTime + "ms");
		}
	}
	/**
	 * 唤起回调
	 */
	private void invokeCallbacks() {
		lock.lock();
		try{
			for (final AsyncRPCCallback callback : pendingCallbacks) {
				runCallback(callback);
			}
		}finally{
			lock.unlock();
		}
		
	}

	private void runCallback(AsyncRPCCallback callback) {
		final RpcResponse response = this.response;
		ClientProxy.submit(new Runnable() {
			
			@Override
			public void run() {
				if (!response.isError()) {
					callback.success(response.getResult());
				}else {
					callback.fail(new RuntimeException("响应错误", new Throwable(response.getError())));
				}
				
			}
		});
		
	}
	/**
	 * 添加回调
	 * @param callback
	 * @return
	 */
	public RPCFuture addCallback(AsyncRPCCallback callback){
		lock.lock();
		try{
			if (isDone()) {
				runCallback(callback);
			}else {
				this.pendingCallbacks.add(callback);
			}
		}finally{
			lock.unlock();
		}
		return this;
	}

	@Override
	public Object get(long timeout, TimeUnit unit) throws InterruptedException,
			ExecutionException, TimeoutException {
		
		//tryAcquireSharedNanos  tryAcquireNanos
		boolean success = sync.tryAcquireNanos(-1, unit.toNanos(timeout));
		if (success) {
			if (this.response != null) {
				return this.response.getResult();
			}else {
				return null;
			}
		}else {
			throw new RuntimeException("超时请求，请求id="+this.request.getRequestId()
					+"请求类名："+this.request.getClassName()
					+"请求方法："+this.request.getMethodName());
		}
	}
	
	
	static class Sync extends AbstractQueuedSynchronizer{

		/**
		 * 
		 */
		private static final long serialVersionUID = 1L;
		//future status
		private final int done = 1;//完成
		private final int pending = 0;//未完成
		
		/**
		 * 获得
		 */
		@Override
		protected boolean tryAcquire(int arg) {
			return getState() == done;
		}
		/**
		 * 释放
		 */
		@Override
		protected boolean tryRelease(int arg) {
			if (getState() == pending) {
				if (compareAndSetState(pending, done)) {
					return true;
				}else {
					return false;
				}
			}else {
				return true;
			}
		}
		@Override
		protected int tryAcquireShared(int acquires) {
			System.out.println(acquires);
			return acquires;
		}
		
		public boolean isDone() {
			getState();
			return getState() == done;
		}
	}
	
}
